package com.ia.aistream.manager.framework.redis.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.google.common.collect.Sets;
import com.ia.aistream.manager.framework.redis.RedisZsetManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.Set;

/**
 * @author chenjunsheng
 */
@Service
public class RedisZsetManagerImpl implements RedisZsetManager {
    @Autowired
    private StringRedisTemplate redisTemplate;

    @Override
    public void zadd(String key, double score, Object member) {
        redisTemplate.boundZSetOps(key).add(JSON.toJSONString(member), score);
    }

    @Override
    public Double zscore(String key, Object member) {
        return redisTemplate.boundZSetOps(key).score(JSON.toJSONString(member));
    }

    @Override
    public Double zincrby(String key, double increment, Object member) {
        return redisTemplate.boundZSetOps(key).incrementScore(JSON.toJSONString(member), increment);
    }

    @Override
    public long zcard(String key) {
        Long card = redisTemplate.boundZSetOps(key).zCard();
        return card == null ? 0 : card;
    }

    @Override
    public <T> Set<T> zrange(String key, int start, int stop) {
        Set<String> set = redisTemplate.boundZSetOps(key).range(start, stop);
        Set<T> result = Sets.newLinkedHashSet();
        if (CollectionUtils.isEmpty(set)) {
            return result;
        }

        TypeReference<T> reference = new TypeReference<T>() {
        };
        set.forEach(string -> result.add(JSON.parseObject(string, reference)));
        return result;
    }

    @Override
    public <T> Set<ZSetOperations.TypedTuple<T>> zrange(String key, int start, int stop, boolean withscores) {
        Set<ZSetOperations.TypedTuple<String>> set = redisTemplate.boundZSetOps(key).rangeWithScores(start, stop);
        Set<ZSetOperations.TypedTuple<T>> result = Sets.newLinkedHashSet();
        if (CollectionUtils.isEmpty(set)) {
            return result;
        }
        TypeReference<T> reference = new TypeReference<T>() {
        };
        set.forEach(tuple -> result.add(new DefaultTypedTuple<T>(JSON.parseObject(tuple.getValue(), reference), withscores ? tuple.getScore() : null)));
        return result;
    }

    @Override
    public <T> Set<T> zrangebyscore(String key, double min, double max) {
        Set<String> set = redisTemplate.opsForZSet().rangeByScore(key, min, max);
        Set<T> result = Sets.newLinkedHashSet();
        if (CollectionUtils.isEmpty(set)) {
            return result;
        }

        TypeReference<T> reference = new TypeReference<T>() {
        };

        set.forEach(string -> result.add(JSON.parseObject(string, reference)));
        return result;
    }

    @Override
    public <T> Set<T> zrangebyscore(String key, double min, double max, int offset, int count) {
        Set<String> set = redisTemplate.opsForZSet().rangeByScore(key, min, max, offset, count);
        Set<T> result = Sets.newLinkedHashSet();
        if (CollectionUtils.isEmpty(set)) {
            return result;
        }

        TypeReference<T> reference = new TypeReference<T>() {
        };

        set.forEach(string -> result.add(JSON.parseObject(string, reference)));
        return result;
    }

    @Override
    public <T> Set<ZSetOperations.TypedTuple<T>> zrangebyscore(String key, double min, double max, boolean withscores, int offset, int count) {
        Set<ZSetOperations.TypedTuple<T>> result = Sets.newLinkedHashSet();

        if (withscores) {
            Set<ZSetOperations.TypedTuple<String>> set = redisTemplate.opsForZSet().rangeByScoreWithScores(key, min, max, offset, count);
            if (CollectionUtils.isEmpty(set)) {
                return result;
            }
            TypeReference<T> reference = new TypeReference<T>() {
            };
            set.forEach(tuple -> result.add(new DefaultTypedTuple<T>(JSON.parseObject(tuple.getValue(), reference), tuple.getScore())));
        } else {
            Set<String> set = redisTemplate.opsForZSet().rangeByScore(key, min, max, offset, count);
            if (CollectionUtils.isEmpty(set)) {
                return result;
            }
            TypeReference<T> reference = new TypeReference<T>() {
            };
            set.forEach(string -> result.add(new DefaultTypedTuple<T>(JSON.parseObject(string, reference), null)));
        }

        return result;
    }

    @Override
    public <T> Set<T> zrevrange(String key, int start, int stop) {
        Set<String> set = redisTemplate.boundZSetOps(key).reverseRange(start, stop);
        Set<T> result = Sets.newLinkedHashSet();
        if (CollectionUtils.isEmpty(set)) {
            return result;
        }

        TypeReference<T> reference = new TypeReference<T>() {
        };
        set.forEach(string -> result.add(JSON.parseObject(string, reference)));
        return result;
    }

    @Override
    public void zrem(String key, Object member) {
        redisTemplate.boundZSetOps(key).remove(JSON.toJSONString(member));
    }
}
